package com.hy.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hy.mapper.TbHotelMapper;
import com.hy.pojo.Hotel;
import com.hy.pojo.HotelRequest;
import com.hy.service.TbHotelService;
import org.apache.commons.lang3.ObjectUtils;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author huangy
 * @since 2025-04-22
 */
@Service
public class TbHotelServiceImpl extends ServiceImpl<TbHotelMapper, Hotel> implements TbHotelService {
    private final RestHighLevelClient esClient;

    public TbHotelServiceImpl(RestHighLevelClient esClient) {
        this.esClient = esClient;
    }

    @Override
    public void syncData() throws Exception {
        // 分页参数
        int pageSize = 500;
        int currentPage = 1;
        long total = 0;

        // 使用SLF4J日志
        Logger logger = LoggerFactory.getLogger(this.getClass());

        try {
            while (true) {
                // 分页查询
                Page<Hotel> page = new Page<>(currentPage, pageSize);
                Page<Hotel> hotelPage = baseMapper.selectPage(page, null);
                List<Hotel> hotels = hotelPage.getRecords();

                if (hotels.isEmpty()) {
                    break;
                }

                // 创建批量请求
                BulkRequest bulkRequest = new BulkRequest();
                for (Hotel hotel : hotels) {
                    // 使用酒店唯一ID作为文档ID
                    IndexRequest request = new IndexRequest("hotel")
                            .id(hotel.getId().toString())
                            .source(JSON.toJSONString(hotel), XContentType.JSON);
                    bulkRequest.add(request);
                }

                // 执行批量操作
                if (bulkRequest.numberOfActions() > 0) {
                    BulkResponse response = esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                    if (response.hasFailures()) {
                        logger.error("ES批量插入失败：{}", response.buildFailureMessage());
                    }
                    total += hotels.size();
                }

                // 退出条件
                if (!hotelPage.hasNext()) {
                    break;
                }
                currentPage++;
            }

            logger.info("数据同步完成，共同步{}条记录", total);
        } catch (IOException e) {
            // 添加重试机制或告警
            logger.error("ES同步发生异常：", e);
            throw new Exception("数据同步失败，请检查ES连接", e);
        }
    }

    @Override
    public void getData() throws IOException {
        // 创建搜索请求对象，指定索引为"hotel"
        SearchRequest request = new SearchRequest("hotel");

        // 1.设置查询条件，使用match_all查询，表示查询所有文档
        request.source().query(QueryBuilders.matchAllQuery());

        int page = 1, size = 10;
        //设置分页和排序
        request.source().from((page - 1) * size).size(10);
        request.source().sort("price", SortOrder.ASC);
        //高亮字段
        request.source().highlighter(new HighlightBuilder()
                .field("name")
                .requireFieldMatch(false));

        // 2.设置查询条件，使用match查询，表示查询所有文档
        // request.source().query(QueryBuilders.matchQuery("starName","石"));
        // 3.bool查询
//        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
//        boolQueryBuilder.must(QueryBuilders.termQuery("city", "上海"));
//        boolQueryBuilder.filter(QueryBuilders.rangeQuery("price").lte(250));
        // 设置查询条件，使用bool查询，表示查询所有文档
//        request.source().query(boolQueryBuilder);

        // 执行搜索请求，并获取搜索响应
        SearchResponse search = esClient.search(request, RequestOptions.DEFAULT);
        //获取高亮响应结果由于不能修改源数据所以不一样
        parseResponse(search);
    }

    /**
     * 统计酒店数据的品牌分布情况。
     * 该方法通过ES聚合查询，统计"hotel"索引中"brand"字段的分布情况，
     * 并限制返回的品牌数量为10个。
     */
    @Override
    public void countData() throws IOException {
        // 构造搜索请求，指定索引为"hotel"
        SearchRequest searchRequest = new SearchRequest("hotel");

        // 设置查询结果大小为0，因为只需要聚合结果，不需要文档数据
        searchRequest.source().size(0);

        // 添加聚合查询，统计"brand"字段的分布情况，限制返回的品牌数量为10个
        searchRequest.source().aggregation(AggregationBuilders.terms("brandAgg")
                .field("brand")
                .size(10));
        SearchResponse search = esClient.search(searchRequest, RequestOptions.DEFAULT);
        for (Aggregation aggregation : search.getAggregations()) {
            // 获取品牌分布聚合结果
            Terms brandAgg = search.getAggregations().get("brandAgg");
            // 获取品牌分布聚合结果中的品牌名称和数量
            for (Terms.Bucket brandBucket : brandAgg.getBuckets()) {
                String brandName = brandBucket.getKeyAsString();
                long brandCount = brandBucket.getDocCount();
                System.out.println("品牌：" + brandName+ "，数量：" + brandCount);
            }
        }
    }

    @Override
    public Map<String, List<String>> filterData() throws IOException {
        // 构造搜索请求，指定索引为"hotel"
        SearchRequest searchRequest = new SearchRequest("hotel");

        // 设置查询结果大小为0，因为只需要聚合结果，不需要文档数据
        searchRequest.source().size(0);

        // 添加聚合查询，统计"brand"字段的分布情况，限制返回的品牌数量为10个
        searchRequest.source().aggregation(AggregationBuilders.terms("brandAgg")
                .field("brand")
                .size(10));
        searchRequest.source().aggregation(AggregationBuilders.terms("cityAgg")
                .field("city")
                .size(10));
        searchRequest.source().aggregation(AggregationBuilders.terms("startNameAgg")
                .field("startName")
                .size(10));
        Map<String, List<String>> map = new HashMap<>();
        SearchResponse search = esClient.search(searchRequest, RequestOptions.DEFAULT);
        for (Aggregation aggregation : search.getAggregations()) {
            // 获取品牌分布聚合结果
            Terms brandAgg = search.getAggregations().get("brandAgg");
            Terms cityAgg = search.getAggregations().get("cityAgg");
            Terms startNameAgg = search.getAggregations().get("startNameAgg");
            List<String> list = new ArrayList<>();

            // 获取品牌分布聚合结果中的品牌名称和数量
            for (Terms.Bucket brandBucket : brandAgg.getBuckets()) {
                String brandName = brandBucket.getKeyAsString();
                list.add(brandName);
                long brandCount = brandBucket.getDocCount();
                System.out.println("品牌：" + brandName + "，数量：" + brandCount);
            }
            map.put("品牌",list);
            List<String> list1 = new ArrayList<>();
            for (Terms.Bucket cityBucket : cityAgg.getBuckets()) {
                String cityName = cityBucket.getKeyAsString();
                list1.add(cityName);
                long cityCount = cityBucket.getDocCount();
                System.out.println("城市：" + cityName + "，数量：" + cityCount);
            }
            map.put("城市",list1);
            List<String> list2 = new ArrayList<>();
            for (Terms.Bucket startNameBucket : startNameAgg.getBuckets()) {
                String startName = startNameBucket.getKeyAsString();
                list2.add(startName);
                long startNameCount = startNameBucket.getDocCount();
                System.out.println("星级：" + startName + "，数量：" + startNameCount);
            }
            map.put("星级",list2);
        }
        return map;
    }

    @Override
    public Map<String, List<String>> filters(HotelRequest request) {
        return null;

    }


    private void parseResponse(SearchResponse search) throws IOException {

        // 获取搜索结果中的命中信息
        SearchHits hits = search.getHits();

        // 获取总命中数
        long value = hits.getTotalHits().value;
        System.out.println("一共搜索到" + value + "条数据");

        // 遍历每个命中的文档
        SearchHit[] hits1 = hits.getHits();
        for (SearchHit documentFields : hits1) {
            // 将文档的源数据转换为JSON字符串
            String sourceAsString = documentFields.getSourceAsString();

            //  1 将JSON字符串解析为Hotel对象
            Hotel hotel = JSON.parseObject(sourceAsString, Hotel.class);
            //2.从文档字段中获取高亮字段，并提取指定字段的高亮片段。
            Map<String, HighlightField> highlightFields =
                    documentFields.getHighlightFields();
            if (ObjectUtils.isNotEmpty(highlightFields)){
                //2.1该函数首先从文档字段中获取所有的高亮字段，然后从高亮字段中提取指定字段（如"name"）的高亮片段，
                HighlightField name = highlightFields.get("name");
                if (ObjectUtils.isEmpty(name)){
                    //   2.2   最后将高亮片段转换为字符串。
                    String string = name.getFragments()[0].string();
                    hotel.setName(string);
                }
            }
            // 1 打印Hotel对象的信息
            System.out.println(hotel.toString());
        }
    }
}
